package zy.learn.demo.structuredstreaming.window

import java.sql.Timestamp

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{OutputMode, Trigger}

/**
 * 输入自定义的时间戳
 */
object WordCountWindow2 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")
    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("WordCountWindow自定义时间戳")
      .getOrCreate()

    import spark.implicits._

    val lines = spark.readStream
      .format("socket") // 设置数据源
      .option("host", "co7-203")
      .option("port", 9999)
      .load
    /* 输入的数据
    * 2020-10-14 09:50:25,hello
    * 2020-10-14 09:50:20,world
    * 输出的结果
+------------------------------------------+-----+-----+
|window                                    |word |count|
+------------------------------------------+-----+-----+
|[2020-10-14 09:50:20, 2020-10-14 09:50:30]|hello|1    |
|[2020-10-14 09:50:25, 2020-10-14 09:50:35]|hello|1    |
+------------------------------------------+-----+-----+
    * */
    val wordsDF = lines.as[String].map(line => {
      val split = line.split(",")
      (split(0), split(1))
    }).toDF("ts", "word")

    import org.apache.spark.sql.functions._

    val wordCounts = wordsDF.groupBy(
      // 调用 window 函数, 返回的是一个 Column 参数 1: df 中表示时间戳的列 参数 2: 窗口长度 参数 3: 滑动步长
      window($"ts", "10 seconds", "5 seconds"),
      $"word"
    ).count().sort("window") // 计数，并按照窗口排序

    val query = wordCounts.writeStream
      .format("console")
      .outputMode(OutputMode.Complete())
      .trigger(Trigger.ProcessingTime(2000))
      .option("truncate", "false")
      .start()

    query.awaitTermination()
  }
}
